package com.ht.api.flink;
//
//import java.util.Properties;
//
//import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.api.common.functions.MapFunction;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.functions.sink.SinkFunction;
//import org.apache.flink.table.api.DataTypes;
//import org.apache.flink.table.api.TableSchema;
//
//import com.starrocks.connector.flink.StarRocksSink;
//import com.starrocks.connector.flink.row.sink.StarRocksSinkRowBuilder;
//import com.starrocks.connector.flink.table.sink.StarRocksSinkOptions;
//import com.ververica.cdc.connectors.mysql.source.MySqlSource;
//import com.ververica.cdc.connectors.mysql.table.StartupOptions;
//import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
//
//import cn.hutool.json.JSONObject;
//import cn.hutool.json.JSONUtil;
//import lombok.AllArgsConstructor;
//import lombok.Data;
//import lombok.extern.slf4j.Slf4j;
//
//@Slf4j
public class MysqlToSelectDBWithSink {
//
//    @Data
//    @AllArgsConstructor
//    public static class User {
//        private Integer id;
//        private String name;
//        private Integer sex;
//        private Integer age;
//    }
//
//    @Data
//    public static class Config {
//        // MySQL配置
//        private String hostname = "127.0.0.1";
//        private String username = "root";
//        private String password = "aixi_2028";
//        private int port = 3306;
//        private String database = "test";
//        private String table = "users";
//
//        // SelectDB配置
//        private String feHost = "selectdb-cn-rp64247ug01-public.selectdbfe.rds.aliyuncs.com";
//        private int jdbcPort = 9030;
//        private int loadPort = 8080;
//        private String dbName = "test";
//        private String tableName = "users";
//        private String selectDBUsername = "admin";
//        private String selectDBPassword = "xx&*(%^Ht20241234yuiexx";
//    }
//
//    public static void main(String[] args) throws Exception {
//        System.setProperty("jdbc.drivers", "com.mysql.cj.jdbc.Driver");
//        Config config = new Config();
//
//        // 1. 设置环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);  // 设置并行度
//
//        // 2. 配置MySQL CDC Source
//        Properties jdbcProperties = new Properties();
//        jdbcProperties.setProperty("useSSL", "false");
//        //jdbcProperties.setProperty("driverClassName", "com.mysql.cj.jdbc.Driver");  // 修改这里
//        //jdbcProperties.setProperty("jdbc.driver", "com.mysql.cj.jdbc.Driver");
//        jdbcProperties.setProperty("connectTimeout", "30000");
//
//        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
//                .hostname(config.getHostname())
//                .port(config.getPort())
//                .databaseList(config.getDatabase())
//                .tableList(config.getDatabase() + "." + config.getTable())
//                .username(config.getUsername())
//                .password(config.getPassword())
//                .deserializer(new JsonDebeziumDeserializationSchema())
//                .startupOptions(StartupOptions.initial())
//                .jdbcProperties(jdbcProperties)
//                .build();
//
//        // 3. 创建CDC数据流并转换数据
//        DataStream<User> userStream = env
//                .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL CDC Source")
//                .map(new MapFunction<String, User>() {
//                    @Override
//                    public User map(String value) throws Exception {
//                        try {
//                            JSONObject json = JSONUtil.parseObj(value);
//                            JSONObject after = json.getJSONObject("after");
//                            String op = json.getStr("op");
//
//                            // 只处理插入和更新操作
//                            if ("c".equals(op) || "u".equals(op)) {
//                                return new User(
//                                        after.getInt("id"),
//                                        after.getStr("name"),
//                                        after.getInt("sex"),
//                                        after.getInt("age")
//                                );
//                            }
//                        } catch (Exception e) {
//                            log.error("解析数据失败: {}", value, e);
//                        }
//                        return null;
//                    }
//                })
//                .filter(user -> user != null);  // 过滤掉解析失败的数据
//
//        // 4. 配置 StarRocks Sink Options
//        StarRocksSinkOptions options = StarRocksSinkOptions.builder()
//                // 使用 9030 端口连接 MySQL 协议
//                .withProperty("jdbc-url", String.format("jdbc:mysql://%s:%d/%s?useSSL=false&connectTimeout=30000&socketTimeout=30000&serverTimezone=UTC", 
//                        config.getFeHost(), 
//                        config.getJdbcPort(),
//                        config.getDbName()))
//                // 使用 8030 端口进行数据导入
//                .withProperty("load-url", String.format("http://%s:%d", config.getFeHost(), config.getLoadPort()))
//                .withProperty("username", config.getSelectDBUsername())
//                .withProperty("password", config.getSelectDBPassword())
//                .withProperty("database-name", config.getDbName())
//                .withProperty("table-name", config.getTableName())
//                // 添加版本相关配置
//                .withProperty("sink.version", "v2")           // 使用 V2 版本
//                .withProperty("sink.semantic", "exactly-once") // 启用事务支持
//                .withProperty("sink.buffer-flush.max-rows", "64000")
//                .withProperty("sink.buffer-flush.interval-ms", "5000")
//                .withProperty("sink.max-retries", "3")
//                .withProperty("sink.properties.format", "json")
//                .withProperty("sink.properties.strip_outer_array", "true")
//                .withProperty("sink.properties.connect_timeout", "30000")
//                .withProperty("sink.properties.socket_timeout", "30000")
//                .build();
//
//        // 5. 创建 StarRocks Sink
//        TableSchema tableSchema = TableSchema.builder()
//                .field("id", DataTypes.INT())
//                .field("name", DataTypes.STRING())
//                .field("sex", DataTypes.INT())
//                .field("age", DataTypes.INT())
//                .build();
//
//        // 6. 创建 StarRocks Sink
//        SinkFunction<User> starRocksSink = StarRocksSink.sink(
//                tableSchema,
//                options,
//                new StarRocksSinkRowBuilder<User>() {
//                    @Override
//                    public void accept(Object[] row, User user) {
//                        row[0] = user.getId();
//                        row[1] = user.getName();
//                        row[2] = user.getSex();
//                        row[3] = user.getAge();
//                    }
//                }
//        );
//
//        // 7. 添加 sink 并执行
//        userStream.addSink(starRocksSink);
//        
//        // 8. 执行任务
//        env.execute("MySQL CDC to SelectDB Sync");
//    }
}
